package com.sojson.util.zookeeper.impl;

import java.util.List;
import java.util.concurrent.CountDownLatch;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.CuratorFrameworkFactory.Builder;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.data.Stat;

import com.sojson.constant.Constant;
import com.sojson.util.StringUtil;
import com.sojson.util.properties.PropertiesUtil;
import com.sojson.util.text.Convert;
import com.sojson.util.zookeeper.CustomWatcher;
import com.sojson.util.zookeeper.ZookeeperUtil;
import com.sojson.util.zookeeper.ZookeeperUtilInterface;
import com.sojson.util.zookeeper.constant.ZooConstant;

import lombok.extern.slf4j.Slf4j;

/**
 * Zookeeper工具类实现类
 *
 * @author liu
 * @date 2020-11-15
 */
@Slf4j
public class CuratorFrameworkImpl implements ZookeeperUtilInterface {

    static CuratorFrameworkImpl zookeeperUtilImpl;
    /** 连接重试之间的初始等待时间(单位: 毫秒,不设置为1秒) */
    public static final int BASE_SLEEP_TIME = Convert.toInt(
        PropertiesUtil.getNumberFormat(Constant.PROPERTIES.getProperty("so.zookeeper.curator.baseSleppTime")), 1000);
    /** 最大连接重试次数(不设置为3) */
    public static final int MAX_RETRIES =
        Convert.toInt(Constant.PROPERTIES.getProperty("so.zookeeper.curator.maxRetries"), 3);
    /** 会话超时时间(单位: 毫秒,不设置为60秒) */
    public static final Integer SESSION_TIMEOUT = Convert.toInt(
        PropertiesUtil.getNumberFormat(Constant.PROPERTIES.getProperty("so.zookeeper.curator.session.timeout")),
        1000 * 60);
    /** 等待连接成功的最大超时时间(单位: 调方法的时候设置,不设置为30秒) */
    public static final Integer CONN_TIMEOUT = Convert.toInt(
        PropertiesUtil.getNumberFormat(Constant.PROPERTIES.getProperty("so.zookeeper.curator.connection.timeout")),
        1000 * 30);
    /** 核心类 */
    CuratorFramework zk;
    /** 构建CuratorFramework的对象 */
    private Builder builder;
    /** 监视器 */
    private CustomWatcher watcher;
    /** Zookeeper服务器地址 */
    private String url;

    public CuratorFrameworkImpl() {}

    public CuratorFrameworkImpl(CustomWatcher watcher) {
        this.watcher = watcher;
    }

    @Override
    protected void finalize() throws Throwable {
        close();
    }

    /**
     * 获取Zookeeper服务器地址
     *
     * @return
     */
    public String getUrl() {
        return url;
    }

    /**
     * 设置Zookeeper服务器地址
     *
     * @param url
     */
    public void setUrl(String url) {
        this.url = url;
    }

    /**
     * 获取ZooKeeper
     *
     * @return
     */
    public CuratorFramework getZk() {
        return zk;
    }

    /**
     * 设置ZooKeeper
     *
     * @param zk
     */
    public void setZk(CuratorFramework zk) {
        this.zk = zk;
    }

    /**
     * 获取构建CuratorFramework的对象
     *
     * @return
     */
    public Builder getBuilder() {
        return builder;
    }

    /**
     * 设置构建CuratorFramework的对象
     *
     * @param builder
     */
    public void setBuilder(Builder builder) {
        this.builder = builder;
    }

    /**
     * 获取Watcher
     *
     * @return
     */
    public Watcher getWatcher() {
        return watcher;
    }

    /**
     * 设置Watcher
     *
     * @param watcher
     */
    public void setWatcher(CustomWatcher watcher) {
        this.watcher = watcher;
    }

    /**
     * 给Zookeeper添加帐号密码
     *
     * @return
     * @throws Exception
     */
    @Override
    public ZookeeperUtilInterface addAcl() {
        return addAcl(ZookeeperUtil.getAclDigestByte());
    }

    /**
     * 给Zookeeper添加帐号密码
     *
     * @param auth 帐号:密码
     * @return
     * @throws Exception
     */
    @Override
    public ZookeeperUtilInterface addAcl(String auth) {
        return addAcl(auth.getBytes());
    }

    /**
     * 给Zookeeper添加帐号密码
     *
     * @param auth 帐号:密码
     * @return
     * @throws Exception
     */
    @Override
    public ZookeeperUtilInterface addAcl(byte[] auth) {
        builder.authorization(ZooConstant.SCHEME_DIGEST, auth);
        return this;
    }

    /**
     * 给Zookeeper添加帐号密码
     *
     * @param username 帐号
     * @param password 密码
     * @return
     * @throws Exception
     */
    @Override
    public ZookeeperUtilInterface addAcl(String username, String password) {
        return addAcl(ZookeeperUtil.getAclDigestByte(username, password));
    }

    /**
     * 同步创建一个持久节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @return
     * @throws Exception
     */
    @Override
    public String create(String path) throws Exception {
        return createByte(path, null);
    }

    /**
     * 同步创建一个持久节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @param data 内容
     * @return
     * @throws Exception
     */
    @Override
    public String create(String path, String data) throws Exception {
        return createByte(path, data.getBytes());
    }

    /**
     * 同步创建一个持久节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @param data 内容
     * @return
     * @throws Exception
     */
    @Override
    public String createByte(String path, byte[] data) throws Exception {
        return zk.create().forPath(path, data);
    }

    /**
     * 同步创建一个持久顺序节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @return
     * @throws Exception
     */
    @Override
    public String createSort(String path) throws Exception {
        return createSortByte(path, null);
    }

    /**
     * 同步创建一个持久顺序节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @param data 内容
     * @return
     * @throws Exception
     */
    @Override
    public String createSort(String path, String data) throws Exception {
        return createSortByte(path, data.getBytes());
    }

    /**
     * 同步创建一个持久顺序节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @param data 内容
     * @return
     * @throws Exception
     */
    @Override
    public String createSortByte(String path, byte[] data) throws Exception {
        return zk.create().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath(path, data);
    }

    /**
     * 同步创建一个有效时间的持久顺序节点
     *
     * @param path      节点
     * @param timeOut   超时时间(单位: 毫秒)
     * @return
     * @throws Exception
     */
    @Override
    public String createSort(String path, long timeOut) throws Exception {
        return createSortByte(path, null, null, timeOut);
    }

    /**
     * 同步创建一个有效时间的持久顺序节点,同时设置节点信息
     *
     * @param path      节点
     * @param stat      节点信息
     * @param timeOut   超时时间(单位: 毫秒)
     * @return
     * @throws Exception
     */
    @Override
    public String createSort(String path, Stat stat, long timeOut) throws Exception {
        return createSortByte(path, null, stat, timeOut);
    }

    /**
     * 同步创建一个有效时间的持久顺序节点
     *
     * @param path      节点
     * @param data   内容
     * @param timeOut   超时时间(单位: 毫秒)
     * @return
     * @throws Exception
     */
    @Override
    public String createSort(String path, String data, long timeOut) throws Exception {
        return createSort(path, data, null, timeOut);
    }

    /**
     * 同步创建一个有效时间的持久顺序节点
     *
     * @param path      节点
     * @param data   内容
     * @param timeOut   超时时间(单位: 毫秒)
     * @return
     * @throws Exception
     */
    @Override
    public String createSortByte(String path, byte[] data, long timeOut) throws Exception {
        return createSortByte(path, data, null, timeOut);
    }

    /**
     * 同步创建一个有效时间的持久顺序节点,同时注册监听
     *
     * @param path      节点
     * @param data   内容
     * @param stat      节点信息
     * @param timeOut   超时时间(单位: 毫秒)
     * @return
     * @throws Exception
     */
    @Override
    public String createSort(String path, String data, Stat stat, long timeOut) throws Exception {
        return createSortByte(path, data.getBytes(), stat, timeOut);
    }

    /**
     * 同步创建一个有效时间的持久顺序节点,同时注册监听
     *
     * @param path      节点
     * @param data   内容
     * @param stat      节点信息
     * @param timeOut   超时时间(单位: 毫秒)
     * @return
     * @throws Exception
     */
    @Override
    public String createSortByte(String path, byte[] data, Stat stat, long timeOut) throws Exception {
        return zk.create().withTtl(timeOut).withMode(CreateMode.PERSISTENT_SEQUENTIAL_WITH_TTL).forPath(path, data);
    }

    /**
     * 同步创建一个临时节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @return
     * @throws Exception
     */
    @Override
    public String createTemporary(String path) throws Exception {
        return createTemporaryByte(path, null);
    }

    /**
     * 同步创建一个临时节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @param data 内容
     * @return
     * @throws Exception
     */
    @Override
    public String createTemporary(String path, String data) throws Exception {
        return createTemporaryByte(path, data.getBytes());
    }

    /**
     * 同步创建一个临时节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @param data 内容
     * @return
     * @throws Exception
     */
    @Override
    public String createTemporaryByte(String path, byte[] data) throws Exception {
        return zk.create().withMode(CreateMode.EPHEMERAL).forPath(path, data);
    }

    /**
     * 同步创建一个临时顺序节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @return
     * @throws Exception
     */
    @Override
    public String createTemporarySort(String path) throws Exception {
        return createTemporarySortByte(path, null);
    }

    /**
     * 同步创建一个临时顺序节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @param data 内容
     * @return
     * @throws Exception
     */
    @Override
    public String createTemporarySort(String path, String data) throws Exception {
        return createTemporarySortByte(path, data.getBytes());
    }

    /**
     * 同步创建一个临时顺序节点,ACL为 world:anyone:cdrwa 等同于如下命令:
     * create path data world:anyone:cdrwa
     *
     * @param path 节点
     * @param data 内容
     * @return
     * @throws Exception
     */
    @Override
    public String createTemporarySortByte(String path, byte[] data) throws Exception {
        return zk.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path, data);
    }

    /**
     * 删除节点
     *
     * @param path 节点
     * @throws Exception
     */
    @Override
    public void delete(String path) throws Exception {
        zk.delete().forPath(path);
    }

    /**
     * 修改节点内容
     *
     * @param path 节点
     * @param data 内容
     * @throws Exception
     */
    @Override
    public void setData(String path, byte[] data) throws Exception {
        zk.setData().forPath(path, data);
    }

    /**
     * 修改节点内容
     *
     * @param path 节点
     * @param data 内容
     * @throws Exception
     */
    @Override
    public void setData(String path, String data) throws Exception {
        setData(path, data.getBytes());
    }

    /**
     * 获取节点内容(不注册监听)
     *
     * @param path 节点
     * @param stat 通过exists查出来的对象
     * @return
     * @throws Exception
     */
    @Override
    public byte[] getData(String path, Stat stat) throws Exception {
        // 不执行Watcher
        return zk.getData().forPath(path);
    }

    /**
     * 获取节点内容(同时注册监听)
     *
     * @param path 节点
     * @param stat 通过exists查出来的对象
     * @return
     * @throws Exception
     */
    @Override
    public byte[] getDataTrue(String path, Stat stat) throws Exception {
        // 执行Watcher
        return zk.getData().usingWatcher(watcher).forPath(path);
    }

    /**
     * 获取节点内容(同时注册监听)
     *
     * @param path 节点
     * @param watcher 监视器
     * @param stat 通过exists查出来的对象
     * @return
     * @throws Exception
     */
    @Override
    public byte[] getData(String path, CustomWatcher watcher, Stat stat) throws Exception {
        // 执行Watcher
        return zk.getData().usingWatcher(watcher).forPath(path);
    }

    /**
     * 获取指定节点的所有子节点(不注册监听)
     *
     * @param path 节点
     * @return
     * @throws Exception
     */
    @Override
    public List<String> getChildren(String path) throws Exception {
        // 不执行Watcher
        return zk.getChildren().forPath(path);
    }

    /**
     * 获取指定节点的所有子节点(同时注册监听)
     *
     * @param path 节点
     * @return
     * @throws Exception
     */
    @Override
    public List<String> getChildrenTrue(String path) throws Exception {
        // 执行Watcher
        return zk.getChildren().usingWatcher(watcher).forPath(path);
    }

    /**
     * 获取指定节点的所有子节点(同时注册监听)
     *
     * @param path 节点
     * @param watcher 监视器
     * @return
     * @throws Exception
     */
    @Override
    public List<String> getChildren(String path, CustomWatcher watcher) throws Exception {
        // 执行Watcher
        return zk.getChildren().usingWatcher(watcher).forPath(path);
    }

    /**
     * 判断节点是否存在(不注册监听)
     *
     * @param path 节点
     * @return
     * @throws Exception
     */
    @Override
    public Stat exists(String path) throws Exception {
        // 不执行Watcher
        return zk.checkExists().forPath(path);
    }

    /**
     * 判断节点是否存在(同时注册监听)
     *
     * @param path 节点
     * @return
     * @throws Exception
     */
    @Override
    public Stat existsTrue(String path) throws Exception {
        // 执行Watcher
        return zk.checkExists().usingWatcher(watcher).forPath(path);
    }

    /**
     * 判断节点是否存在(同时注册监听)
     *
     * @param path 节点
     * @param watcher 监视器
     * @return
     * @throws Exception
     */
    @Override
    public Stat exists(String path, CustomWatcher watcher) throws Exception {
        // 执行Watcher
        return zk.checkExists().usingWatcher(watcher).forPath(path);
    }

    /**
     * 打开连接
     *
     * @return
     */
    @Override
    public CuratorFrameworkImpl open() {
        // 创建一个简单的连接(不能设置ACL)
        // this.zk = CuratorFrameworkFactory.newClient(url, SESSION_TIMEOUT, CONN_TIMEOUT, getRetryPolicy());

        // 创建连接
        zk = builder.build();

        // 打开链接
        zk.start();
        return this;
    }

    /**
     * 获取ZooKeeper(单例)
     *
     * @return
     */
    public static CuratorFrameworkImpl getInstancesSingle() {
        if (isBlankObject(zookeeperUtilImpl)) {
            zookeeperUtilImpl = createInstances(ZookeeperUtil.URL, null, null);
        }
        return zookeeperUtilImpl;
    }

    /**
     * 获取ZooKeeper
     *
     * @return
     */
    @Override
    public CuratorFrameworkImpl getInstances() {
        return getInstances(ZookeeperUtil.URL);
    }

    /**
     * 获取ZooKeeper
     *
     * @param url 连接地址
     * @return
     */
    @Override
    public CuratorFrameworkImpl getInstances(String url) {
        return getInstances(url, null, null);
    }

    /**
     * 获取ZooKeeper
     *
     * @param watcher 监视器
     * @param countDownLatch 因为连接需要时间,所以需要在连接的时候暂停线程,连接成功后会有个回调方法,在回调方法中让计数减一后变成0恢复线程
     * @return
     */
    @Override
    public CuratorFrameworkImpl getInstances(CustomWatcher watcher, CountDownLatch countDownLatch) {
        return getInstances(ZookeeperUtil.URL, watcher, countDownLatch);
    }

    /**
     * 获取ZooKeeper
     *
     * @param url 连接地址
     * @param watcher 监视器
     * @param countDownLatch 因为连接需要时间,所以需要在连接的时候暂停线程,连接成功后会有个回调方法,在回调方法中让计数减一后变成0恢复线程
     * @return
     */
    @Override
    public CuratorFrameworkImpl getInstances(String url, CustomWatcher watcher, CountDownLatch countDownLatch) {
        return createInstances(url, watcher, countDownLatch);
    }

    /**
     * 获取ZooKeeper
     *
     * @param url 连接地址
     * @param watcher 监视器
     * @param countDownLatch 因为连接需要时间,所以需要在连接的时候暂停线程,连接成功后会有个回调方法,在回调方法中让计数减一后变成0恢复线程
     * @return
     */
    private static CuratorFrameworkImpl createInstances(String url, CustomWatcher watcher,
        CountDownLatch countDownLatch) {
        CuratorFrameworkImpl curatorFrameworkImpl = new CuratorFrameworkImpl();
        curatorFrameworkImpl.setWatcher(watcher);
        Builder builder = CuratorFrameworkFactory.builder().connectString(url).retryPolicy(getRetryPolicy())
            .connectionTimeoutMs(CONN_TIMEOUT).sessionTimeoutMs(SESSION_TIMEOUT);
        curatorFrameworkImpl.setBuilder(builder);

        return curatorFrameworkImpl;
    }

    /**
     * 获取连接配置
     *
     * @return
     */
    public static RetryPolicy getRetryPolicy() {
        return new ExponentialBackoffRetry(BASE_SLEEP_TIME, MAX_RETRIES);
    }

    public String createEphemeralSequential(CuratorFramework client, String path, byte[] payload) throws Exception {
        // this will create the given EPHEMERAL-SEQUENTIAL ZNode with the given
        // data using Curator protection.
        return client.create().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path, payload);
    }

    public void setDataAsync(CuratorFramework client, String path, byte[] payload) throws Exception {
        // this is one method of getting event/async notifications
        CuratorListener listener = new CuratorListener() {
            @Override
            public void eventReceived(CuratorFramework client, CuratorEvent event) throws Exception {
                // examine event for details
            }
        };
        client.getCuratorListenable().addListener(listener);
        // set data for the given node asynchronously. The completion
        // notification
        // is done via the CuratorListener.
        client.setData().inBackground().forPath(path, payload);
    }

    public void setDataAsyncWithCallback(CuratorFramework client, BackgroundCallback callback, String path,
        byte[] payload) throws Exception {
        // this is another method of getting notification of an async completion
        client.setData().inBackground(callback).forPath(path, payload);
    }

    public void guaranteedDelete(CuratorFramework client, String path) throws Exception {
        // delete the given node and guarantee that it completes
        client.delete().guaranteed().forPath(path);
    }

    public List<String> watchedGetChildren(CuratorFramework client, String path) throws Exception {
        /**
         * Get children and set a watcher on the node. The watcher notification
         * will come through the CuratorListener (see setDataAsync() above).
         */
        return client.getChildren().watched().forPath(path);
    }

    /**
     * 关闭连接
     *
     * @return
     * @throws Exception
     */
    @Override
    public void close() throws Exception {
        log.info("正在关闭Zookeeper");
        if (zk != null) {
            try {
                zk.close();
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
            zk = null;
        }
    }

    /**
     * 添加监听(只监听本身)
     *
     * @return
     * @throws Exception
     */
    public NodeCache addNodeCache(String path, CustomWatcher watcher) throws Exception {
        final NodeCache nodeCache = new NodeCache(zk, path, false);
        nodeCache.start(true);
        nodeCache.getListenable().addListener(new NodeCacheListener() {

            @Override
            public void nodeChanged() throws Exception {
                watcher.onNodeUpdated(zk, nodeCache.getCurrentData());
            }

        });
        return nodeCache;
    }

    /**
     * 添加监听(只监听子节点)
     *
     * @return
     * @throws Exception
     */
    public PathChildrenCache addPathChildrenWatcher(String path, CustomWatcher watcher) throws Exception {
        final PathChildrenCache pathChildrenCache = new PathChildrenCache(zk, path, true);
        pathChildrenCache.start(StartMode.BUILD_INITIAL_CACHE);
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {

            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                ChildData data = event.getData();
                switch (event.getType()) {
                    case CHILD_ADDED:
                        watcher.onChildAdd(client, data);
                        break;
                    case CHILD_REMOVED:
                        watcher.onChildRemove(client, data);
                        break;
                    case CHILD_UPDATED:
                        watcher.onChildUpdate(client, data);
                        break;
                    default:
                        break;
                }
            }

        });
        return pathChildrenCache;
    }

    /**
     * 添加监听(监听本身和子节点)
     *
     * @return
     * @throws Exception
     */
    public TreeCache addTreeWatcher(String path, CustomWatcher watcher) throws Exception {
        final TreeCache treeCache = new TreeCache(zk, path);
        treeCache.start();
        treeCache.getListenable().addListener(new TreeCacheListener() {

            @Override
            public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
                ChildData data = event.getData();
                switch (event.getType()) {
                    case NODE_ADDED:
                        watcher.onTreeNodeAdd(client, data);
                        break;
                    case NODE_REMOVED:
                        watcher.onTreeNodeRemove(client, data);
                        break;
                    case NODE_UPDATED:
                        watcher.onTreeNodeUpdated(client, data);
                        break;
                    default:
                        break;
                }
            }

        });
        return treeCache;
    }

    /**
     * 删除监听
     *
     * @return
     * @throws Exception
     */
    public NodeCache removeNodeCache(NodeCache watcher) throws Exception {
        // TODO
        // final NodeCache nodeCache = new NodeCache(zk, path, false);
        // nodeCache.start(true);
        // nodeCache.getListenable().removeListener(watcher.);
        return null;
    }

    /**
     * 删除监听
     *
     * @return
     * @throws Exception
     */
    public PathChildrenCache removePathChildrenWatcher(String path, CustomWatcher watcher) throws Exception {
        // TODO
        final PathChildrenCache pathChildrenCache = new PathChildrenCache(zk, path, true);
        pathChildrenCache.start(StartMode.BUILD_INITIAL_CACHE);
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {

            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                ChildData data = event.getData();
                switch (event.getType()) {
                    case CHILD_ADDED:
                        watcher.onChildAdd(client, data);
                        break;
                    case CHILD_REMOVED:
                        watcher.onChildRemove(client, data);
                        break;
                    case CHILD_UPDATED:
                        watcher.onChildUpdate(client, data);
                        break;
                    default:
                        break;
                }
            }

        });
        return pathChildrenCache;
    }

    /**
     * 删除监听
     *
     * @return
     * @throws Exception
     */
    public TreeCache removeTreeWatcher(String path, CustomWatcher watcher) throws Exception {
        // TODO
        final TreeCache treeCache = new TreeCache(zk, path);
        treeCache.start();
        treeCache.getListenable().addListener(new TreeCacheListener() {

            @Override
            public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
                ChildData data = event.getData();
                switch (event.getType()) {
                    case NODE_ADDED:
                        watcher.onTreeNodeAdd(client, data);
                        break;
                    case NODE_REMOVED:
                        watcher.onTreeNodeRemove(client, data);
                        break;
                    case NODE_UPDATED:
                        watcher.onTreeNodeUpdated(client, data);
                        break;
                    default:
                        break;
                }
            }

        });
        return treeCache;
    }

    /**
     * 判断一个对象是否是空(空对象)
     *
     * @param obj
     * @return
     */
    private static boolean isBlankObject(Object obj) {
        return StringUtil.isBlankObject(obj);
    }

}